package wsmsgsubpub

import (
	"fmt"
	"gitee.com/ymofen/gobase"
	"gitee.com/ymofen/gobase/gojson"
	"gitee.com/ymofen/supervalue"
	"golang.org/x/net/websocket"
	"sync"
	"sync/atomic"
	"time"
)

const (
	WS_HTTP_CHANNEL_CMD     string = "/cmd"
	WS_TOPIC_KEY            string = "__topic"
	WS_TOKEN_KEY            string = "__token"
	WS_HTTP_CMD_subscribe          = "subscribe"
	WS_HTTP_CMD_unsubscribe        = "unsubscribe"
	WS_MSG_KEY                     = "__msg"

	WS_HTTP_CHANNEL_EVENT_ONCONNECTED string = "/onconnected"
)

type WsJsonMsgToken struct {
	token   string
	wait    chan struct{}
	respBuf []byte
}

type WsJsonMsg struct {
	Topic string
	Msg   []byte
}

var (
	tokenPrefix string = gobase.RandKeyString(8)
	tokenSN     int32  = 0
)

func NewWsJsonMsgToken() *WsJsonMsgToken {
	rval := &WsJsonMsgToken{token: fmt.Sprintf("%s-%d", tokenPrefix, atomic.AddInt32(&tokenSN, 1))}
	rval.wait = make(chan struct{}, 1)
	return rval
}

func (this *WsJsonMsgToken) Done() bool {
	select {
	case this.wait <- struct{}{}:
		return true
	default:
		return false
	}
}

func (this *WsJsonMsgToken) Wait(t time.Duration) bool {
	select {
	case <-this.wait:
		return true
	case <-time.After(t):
		return false
	}
}

type WsJsonMsgCli struct {
	connectDNA   int32
	WaitRespFlag int8
	wscli        *WsClient
	respTimeOut  time.Duration
	cmdLk        sync.RWMutex

	subLk       sync.RWMutex
	subMap      map[string]int32
	cacheCmd    map[string]*WsJsonMsgToken
	Msglst      chan *WsJsonMsg
	OnConnected func(sender *WsJsonMsgCli)
}

func NewWsJsonMsgCli(url string) *WsJsonMsgCli {
	rval := &WsJsonMsgCli{cacheCmd: make(map[string]*WsJsonMsgToken), subMap: make(map[string]int32), WaitRespFlag: 1}
	rval.respTimeOut = time.Second * 5
	rval.Msglst = make(chan *WsJsonMsg)
	rval.wscli = NewWsClient()
	rval.wscli.OnConnected = rval.onWsCliConnected
	rval.wscli.ConfigWsReqUrl(url, "")
	rval.wscli.OnRecv = rval.onRecv
	return rval
}

func (this *WsJsonMsgCli) Open() {
	if this.wscli.CheckStart() {
		time.Sleep(time.Millisecond * 200)
	}
}

func (this *WsJsonMsgCli) Close() error {
	this.wscli.Stop()
	this.wscli.WaitFor(time.Second * 5)
	close(this.Msglst)
	return nil
}

func (this *WsJsonMsgCli) onWsCliConnected(sender *WsClient) {
	dna := atomic.AddInt32(&this.connectDNA, 1)
	evt := this.OnConnected
	if evt != nil {
		evt(this)
	}
	go this.resubAll(dna)
}

func (this *WsJsonMsgCli) RecvMsg() *WsJsonMsg {
	return <-this.Msglst
}

func (this *WsJsonMsgCli) onRecv(cli *WsClient, msgtype byte, buf []byte) {
	json, _ := wrapper.NewSVFromBuf(buf)
	if json == nil {
		return
	}
	token := json.StringByName(WS_TOKEN_KEY, "")
	if len(token) > 0 {
		this.cmdLk.RLock()
		cmd := this.cacheCmd[token]
		this.cmdLk.RUnlock()
		if cmd != nil {
			cmd.respBuf = buf
			cmd.Done()
		}
	} else {
		msg := &WsJsonMsg{Topic: json.StringByName(WS_TOPIC_KEY, ""), Msg: []byte(json.StringByName(WS_MSG_KEY, ""))}
		this.Msglst <- msg
	}
}

func (this *WsJsonMsgCli) innerSub(topic string) error {
	token := NewWsJsonMsgToken()
	s := gojson.BuildJSONString(WS_TOPIC_KEY, WS_HTTP_CHANNEL_CMD, "cmd", WS_HTTP_CMD_subscribe, "channels", topic, WS_TOKEN_KEY, token.token)
	return this.sendCmd(s, token)
}

func (this *WsJsonMsgCli) innerUnSub(topic string) error {
	token := NewWsJsonMsgToken()
	s := gojson.BuildJSONString(WS_TOPIC_KEY, WS_HTTP_CHANNEL_CMD, "cmd", WS_HTTP_CMD_unsubscribe, "channels", topic, WS_TOKEN_KEY, token.token)
	return this.sendCmd(s, token)
}

func (this *WsJsonMsgCli) resubAll(dna int32) {
	if gobase.GoFunCatchException {
		defer gobase.DeferCatchPanic()
	}
	this.subLk.Lock()
	defer this.subLk.Unlock()
	for k, v := range this.subMap {
		if dna != atomic.LoadInt32(&this.connectDNA) {
			break // 重连了
		}
		if v != dna {
			if v == -1 {
				delete(this.subMap, k)
			} else {
				if this.innerSub(k) != nil {
					this.subMap[k] = dna
				}
			}
		}
	}
}

func (this *WsJsonMsgCli) Sub(topic string) {
	this.subLk.Lock()
	defer this.subLk.Unlock()
	if _, ok := this.subMap[topic]; !ok {
		if this.innerSub(topic) == nil {
			this.subMap[topic] = this.connectDNA
		} else {
			this.subMap[topic] = 0
		}
	}
}

func (this *WsJsonMsgCli) UnSub(topic string) {
	this.subLk.Lock()
	defer this.subLk.Unlock()
	delete(this.subMap, topic)
	this.innerUnSub(topic)
}

func (this *WsJsonMsgCli) Pub(topic string, msg string) (n int, err error) {
	token := NewWsJsonMsgToken()
	s := gojson.BuildJSONString(WS_TOPIC_KEY, topic, WS_MSG_KEY, msg, WS_TOKEN_KEY, token.token)
	err = this.sendCmd(s, token)
	if err != nil {
		return -1, err
	} else {
		json, _ := wrapper.NewSVFromBuf(token.respBuf)
		n = json.IntByName("n", 0)
		return n, nil
	}
}

func (this *WsJsonMsgCli) sendCmd(s string, token *WsJsonMsgToken) (err error) {
	this.cmdLk.Lock()
	this.cacheCmd[token.token] = token
	this.cmdLk.Unlock()
	defer func() {
		this.cmdLk.Lock()
		delete(this.cacheCmd, token.token)
		this.cmdLk.Unlock()
	}()
	err = this.wscli.SendMsg(websocket.TextFrame, []byte(s))
	if err != nil {
		return err
	}
	if this.WaitRespFlag == 1 {
		if token.Wait(this.respTimeOut) {
			return nil
		} else {
			return fmt.Errorf("wait resp timeout")
		}
	} else {
		return nil
	}
}
